package com.shujiadw

import java.awt.geom.Point2D
import java.text.SimpleDateFormat
import java.util.Date

import com.shujia.common.grid.Grid
import org.apache.spark.rdd.RDD
import org.apache.spark.sql._

object StayPointJob {
  def main(args: Array[String]): Unit = {

    if (args.length == 0) {
      println("请指定时间参数")
      return
    }

    var day_id: String = args(0)

    if ("%y%m%d".equals(day_id)) {
      val f: SimpleDateFormat = new SimpleDateFormat("yyyyMMdd")
      day_id = f.format(new Date)
    }
    println("当前分区时间：" + day_id)


    val spark: SparkSession = SparkSession
      .builder()
      .appName("MergeLocationJob")
      .enableHiveSupport() //使用hived的元数据
      .getOrCreate()

    import spark.implicits._

    /**
      * mdn string comment '手机号码'
      * ,start_time string comment '业务时间'
      * ,county_id string comment '区县编码'
      * ,longi string comment '经度'
      * ,lati string comment '纬度'
      * ,bsid string comment '基站标识'
      * ,grid_id string comment '网格号'
      * ,biz_type string comment '业务类型'
      * ,event_type string comment '事件类型'
      * ,data_source string comment '数据源'
      *
      */


    val merge: Dataset[Row] = spark.table("dwi.dwi_res_regn_mergelocation_msk_d")
      .where($"day_id" === day_id)


    //取出所需要的字段
    val mergeRDD: RDD[(String, String)] = merge.map(row => {
      val mdn: String = row.getAs[String]("mdn")
      val start_time: String = row.getAs[String]("start_time")
      val county_id: String = row.getAs[String]("county_id")
      val grid_id: String = row.getAs[String]("grid_id")

      //以手机号和网格编号区县编号作为key
      (mdn + "_" + grid_id + "_" + county_id, start_time)
    }).rdd

    //安装手机好和网格编号和区县编号分组分组
    val groupByRDD: RDD[(String, Iterable[String])] = mergeRDD.groupByKey()

    /**
      * 合并同一个人在同一个网格中的多个点吗，
      * 取第一个点的开始时间和最后一个点的结束时间
      *
      */

    val stayPoint: DataFrame = groupByRDD.map(kv => {

      val mdnAndGridAndCounty: String = kv._1

      val split: Array[String] = mdnAndGridAndCounty.split("_")
      val mdn: String = split(0)
      val grid: String = split(1)
      val county_id: String = split(2)

      //一个人在一个网格中所有的点
      val points: List[String] = kv._2.toList


      //对所有点进行排序
      val sortPoint: List[String] = points.sortBy(time => time.split(",")(0))


      //取第一个点的开始时间
      val grid_first_time: String = sortPoint.head.split(",")(0)

      //取最后一个点的结束时间
      val grid_last_time: String = sortPoint.last.split(",")(1)


      val format: SimpleDateFormat = new SimpleDateFormat("yyyyMMddHHmmss")

      val startDate: Date = format.parse(grid_first_time)
      val endDate: Date = format.parse(grid_last_time)

      //停留时间，分钟级别
      val duration: Long = Math.abs((endDate.getTime - startDate.getTime) / 60000)


      //获取网格中心点的经纬度
      val p: Point2D.Double = Grid.getCenter(grid.toLong)

      val longi: String = p.getX.formatted("%.4f")
      val lati: String = p.getY.formatted("%.4f")


      (mdn, longi, lati, grid, county_id, duration, grid_first_time, grid_last_time)
    }).toDF()

    //保存数据
    stayPoint.write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dwi/dwi_staypoint_msk_d/day_id=$day_id")


  }
}
